#include "msgqueue_util.h"
#include <errno.h>
#include <fcntl.h>
//#include <stdarg.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/time.h>

#ifdef __linux__
#include <sys/ipc.h>
#include <sys/msg.h>
#else
#include <windows.h>
#endif



#define MAX_MQ_DESCRIPTORS (32)  // 最大消息队列描述符数量
#define DATA_BUFFER_MAXSIZE (8 * 1024) // 消息队列数据单包最大值
#define MQ_PATH_LEN (128) // 路径信息存储空间大小

// 函数声明
static int create_or_open_mq_by_key(int key, int tag);
static int create_or_open_mq(const char* mq_path, int mq_proj_id, int tag);


// 消息队列描述符结构
typedef struct _mq_descriptor{
    int handle; // 通信句柄，通常是在windows下使用，-1为初始无效值
    int msgid; // 消息队列fd，-1为初始无效值
    int valid; // 有效标志 (1=有效, 0=无效)
    char path[MQ_PATH_LEN]; // 路径信息
} mq_descriptor;

// 全局消息队列描述符表，及其控制锁
//static pthread_mutex_t mq_des_lock = PTHREAD_RWLOCK_INITIALIZER;
static pthread_rwlock_t mq_des_rwlock = PTHREAD_RWLOCK_INITIALIZER;
static mq_descriptor mq_descriptors[MAX_MQ_DESCRIPTORS] = {{0}};

// 数据自动读取的回调和调用上下文，线程
static pfn_mq_readdata_callback mq_readdata_callback_pfn = NULL;
static void* mq_call_context = NULL;

// 数据读取的线程相关变量
static int mq_readdata_thread_runtag = 0; // 运行标识，0 -- 未运行，1 -- 运行中
static pthread_t mq_readdata_thread_hd = 0; // 线程句柄

/******************************************************************
 * 功能：初始化描述符表，带有 __attribute__((constructor)) 的函数会在main函数执行之前自动调用
 * ****************************************************************/
static void __attribute__((constructor)) init_mq_table()
{
    pthread_rwlock_wrlock(&mq_des_rwlock);
    for (int i = 0; i < MAX_MQ_DESCRIPTORS; ++i) {
        mq_descriptors[i].handle = -1;
        mq_descriptors[i].msgid = -1;
        mq_descriptors[i].valid = 0;
        memset(mq_descriptors[i].path, 0, MQ_PATH_LEN);
    }
    pthread_rwlock_unlock(&mq_des_rwlock);
    return;
}

/******************************************************************
 * 功能：获取空闲描述符索引
 * 返回值：
 *      int：空闲描述符的索引，-1 -- 无效，0到（MAX_MQ_DESCRIPTORS-1）有效的索引
 * ****************************************************************/
static int get_free_descriptor() {
    int index = -1; // 无可用描述符
    pthread_rwlock_rdlock(&mq_des_rwlock);
    for (int i = 0; i < MAX_MQ_DESCRIPTORS; ++i) {
        if (!(mq_descriptors[i].valid)) {
            index = i;
            break;
        }
    }
    pthread_rwlock_unlock(&mq_des_rwlock);
    return index;
}

/******************************************************************
 * 功能：根据消息队列的id获取索引
 * 输入参数：
 *      fd：消息队列的fd
 * 返回值：
 *      int：获取到的索引，-1 -- 无效，0到（MAX_MQ_DESCRIPTORS-1）有效的索引
 * ****************************************************************/
static int get_mq_index(int fd) {
    int index = -1; // 无可用描述符
    pthread_rwlock_rdlock(&mq_des_rwlock);
    for (int i = 0; i < MAX_MQ_DESCRIPTORS; ++i) {
        if (fd == mq_descriptors[i].msgid && 1 == mq_descriptors[i].valid) {
            index = i;
            break;
        }
    }
    pthread_rwlock_unlock(&mq_des_rwlock);
    return index;
}

/******************************************************************
 * 功能：创建或者打开一个消息队列
 * 输入参数：
 *      key：通信双方约定的相同键值信息
 *      tag：操作标志，0 -- 创建并打开，其余值 -- 打开已存在的
 * 返回值：
 *      int：消息队列的fd信息，小于0 -- 失败，大于等于0 -- 成功
 * ****************************************************************/
static int create_or_open_mq_by_key(int key, int tag)
{
    if (0 >= key) {
        printf("create_or_open_mq_by_key: input key (%d) error, must bigger than 0.\r\n", key);
        return -1;
    }

#ifdef __linux__
    // 先尝试获取index，避免消息队列描述符索引被用完
    int idx = get_free_descriptor();
    if (-1 == idx) {
        errno = EMFILE; // 描述符耗尽
        perror("create_or_open_mq: there is no more available descriptor (max size = 32), please close and destroy some message queue.\r\n");
        return -1;
    }

    int fd = -1;
    if (0 == tag) {
        // 创建并打开消息队列
        fd = msgget(key, 0666 | IPC_CREAT); // 允许复用
        //fd = msgget(key, 0666 | IPC_CREAT | IPC_EXCL); // 不允许复用
    } else {
        // 打开一个已存在的消息队列
        fd = msgget(key, 0666);
    }

    // msgget调用成功是，返回一个非负整数（0-0x7FFFFFFF）
    // msgget返回值0是合法的队列标识符，但极少见（因为大多数系统会跳过0作为有效ID）
    if (0 > fd) {
        perror("create_or_open_mq: creat or open messagequeue call msgget failed.\r\n");
        return -1;
    }

    // 更新信息
    pthread_rwlock_wrlock(&mq_des_rwlock);
    mq_descriptors[idx].handle = fd;
    mq_descriptors[idx].msgid = fd;
    mq_descriptors[idx].valid = 1;
    pthread_rwlock_unlock(&mq_des_rwlock);

    return fd; // 返回有效的消息队列的id
#else
    char mq_path[32] = {0};
    snprintf(mq_path, sizeof (mq_path), "mq_win_%d", key);
    return create_or_open_mq(mq_path, key, tag);
#endif
}

/******************************************************************
 * 功能：创建或者打开一个消息队列
 * 输入参数：
 *      mq_path：通信双方约定的相同路径信息，路径未存在时主动创建，销毁队列时删除路径
 *              注：路径需要有读写访问权限
 *      mq_proj_id：通信双方约定的key信息，通常为一个int32的数
 *      tag：操作标志，0 -- 创建并打开，其余值 -- 打开已存在的
 * 返回值：
 *      int：消息队列的fd信息，小于0 -- 失败，大于等于0 -- 成功
 * ****************************************************************/
static int create_or_open_mq(const char* mq_path, int mq_proj_id, int tag)
{
    if (!mq_path) {
        perror("create_or_open_mq: input message queue path is nullptr.\r\n");
        return -1;
    }

    if (MQ_PATH_LEN <= strlen(mq_path)) {
        printf("create_or_open_mq: input path error, too long (%ld), must less (%d).\r\n", strlen(mq_path), MQ_PATH_LEN);
        return -1;
    }

#ifdef __linux__
    if (0 == tag) {
        // 创建时才创建通信的路径
        int fd = open(mq_path, O_RDWR | O_CREAT | O_TRUNC, 0666);
        if (0 > fd) {
            printf("create_or_open_mq: create path: %s failed.\r\n", mq_path);
            return -1;
        }
        close(fd); // 路径创建成功后关闭文件句柄
    }

    key_t key = ftok(mq_path, mq_proj_id); // 生成唯一键值，正常情况下返回有效范围1-255，0比较少见
    if (0 > key) {
        perror("create_or_open_mq: ftok failed.\r\n");
        return -1;
    }

    int fd = create_or_open_mq_by_key((int)key, tag);
    if (0 <= fd) {
        // 更新信息，其余信息在create_or_open_mq_by_key中已经更新
        int idx = get_mq_index(fd);
        if (0 <= idx && MAX_MQ_DESCRIPTORS > idx) {
            pthread_rwlock_wrlock(&mq_des_rwlock);
            memcpy(mq_descriptors[idx].path, mq_path, strlen(mq_path));
            pthread_rwlock_unlock(&mq_des_rwlock);
        }
    }

    return fd; // 返回有效的消息队列的id
#else
    // windows下没有合适现成的消息队列，这里采用共享内存实现
    // 先尝试获取index，避免消息队列描述符索引被用完
    int idx = get_free_descriptor();
    if (-1 == idx) {
        errno = EMFILE; // 描述符耗尽
        perror("create_or_open_mq: there is no more available descriptor (max size = 32), please close and destroy some message queue.\r\n");
        return -1;
    }

    // 共享内存数据，头部4个字节为数据长度
    HANDLE hMapFile = NULL;
    if (0 == tag) {
        // 创建并打开
        // 创建共享内存
        hMapFile = CreateFileMapping(INVALID_HANDLE_VALUE // 使用系统分页文件
                                     , NULL// 默认安全属性
                                     , PAGE_READWRITE // 读写权限
                                     , 0 // 最大大小高字序部分
                                     , DATA_BUFFER_MAXSIZE // 最大大小低字序部分
                                     , mq_path // 共享内存名称
                                     );

        if (NULL == hMapFile) {
            printf("create_or_open_mq: CreateFileMapping failed, error code (%lu).\r\n", GetLastError());
            return -1;
        }


    } else {
        // 只打开
        // 连接共享内存
        hMapFile = OpenFileMapping(FILE_MAP_ALL_ACCESS // 读写权限
                                   , FALSE // 不继承句柄
                                   , mq_path // 共享内存名称
                                   );

        if (NULL == hMapFile) {
            printf("create_or_open_mq: OpenFileMapping failed, error code (%lu).\r\n", GetLastError());
            return -1;
        }
    }

    // 映射共享内存到进程地址空间
    LPVOID hBuffer = MapViewOfFile(hMapFile // 共享内存对象
                                   , FILE_MAP_ALL_ACCESS // 读写权限
                                   , 0 // 偏移高字序部分
                                   , 0 // 偏移低字序部分
                                   , DATA_BUFFER_MAXSIZE // 映射大小
                                   );

    if (NULL == hBuffer) {
        printf("create_or_open_mq: MapViewOfFile failed, error code (%lu).\r\n", GetLastError());
        CloseHandle(hMapFile);
        return -1;
    }

    pthread_rwlock_wrlock(&mq_des_rwlock);
    mq_descriptors[idx].handle = (int)hMapFile;
    mq_descriptors[idx].msgid = (int)hBuffer;
    mq_descriptors[idx].valid = 1;
    memcpy(mq_descriptors[idx].path, mq_path, strlen(mq_path));
    pthread_rwlock_unlock(&mq_des_rwlock);

    return (int)hBuffer; // 返回有效的消息队列的id
#endif
}

/******************************************************************
 * 功能：关闭或销毁消息队列
 * 输入参数：
 *      fd：消息队列的fd
 *      tag：操作标志，0 -- 关闭并销毁，其余值 -- 关闭已存在的
 * ****************************************************************/
static void close_or_destroy_mq(int fd, int tag)
{
    // 先检测消息队列id的有效性
    int idx = get_mq_index(fd);
    if (-1 == idx || MAX_MQ_DESCRIPTORS <= idx) {
        printf("close_or_destroy_mq: invaild message id (%d).\r\n", fd);
        return;
    }

#ifdef __linux__
    if (0 == tag) {
        // 销毁消息队列
        if (0 > msgctl(fd, IPC_RMID, NULL)) {
            perror("close_or_destroy_mq: msgctl failed.\r\n");
            return;
        }

        // 删除消息队列依赖的文件，如存在
        pthread_rwlock_rdlock(&mq_des_rwlock);
        if (0 < strlen(mq_descriptors[idx].path)) {
            remove(mq_descriptors[idx].path);
        }
        pthread_rwlock_unlock(&mq_des_rwlock);
    }
#else
    LPVOID hBuffer = (LPVOID)fd;
    pthread_rwlock_rdlock(&mq_des_rwlock);
    HANDLE hMapFile = (HANDLE)(mq_descriptors[idx].handle);
    pthread_rwlock_unlock(&mq_des_rwlock);

    // 先要清掉共享内存的数据，避免读取到重复数据
    DWORD dwLen = 0;
    CopyMemory(hBuffer, &dwLen, sizeof(DWORD)); // 将数据长度设置为0
    char* hDataHeader = (char*)hBuffer + sizeof(DWORD); // 获取数据起始位置
    ZeroMemory(hDataHeader, DATA_BUFFER_MAXSIZE - sizeof(DWORD)); // 清空数据区域

    UnmapViewOfFile(hBuffer);
    CloseHandle(hMapFile);
#endif

    // 设置非有效标志
    pthread_rwlock_wrlock(&mq_des_rwlock);
    mq_descriptors[idx].handle = -1;
    mq_descriptors[idx].msgid = -1;
    mq_descriptors[idx].valid = 0;
    memset(mq_descriptors[idx].path, 0, MQ_PATH_LEN);
    pthread_rwlock_unlock(&mq_des_rwlock);

    return;
}

/***************************************************************
 * 功能：select模式下精确睡眠
 * 输入参数：
 *      mill_sec：睡眠时间，毫秒值
 * 返回值：
 *      int：结果，0 -- 成功，<0 -- 失败
 * *************************************************************/
static int select_sleep(int mill_sec)
{
#ifdef __linux__
    struct timeval timeout = {0, mill_sec * 1000};   //{秒, 微秒}
    select(0, NULL, NULL, NULL, &timeout);
#else
    Sleep(mill_sec);
#endif
    return 0;
}

/***************************************************************
 * 功能：消息队列自动读取数据的线程执行函数
 * 输入参数：
 *      void*：线程输入信息，如：这里可以为线程的id
 * 返回值：
 *      void*：线程执行反馈的信息
 * *************************************************************/
static void* pfn_readdata_mq(void* arg)
{
    pthread_setname_np(pthread_self(), "MsgQueueRead"); // 设置线程名称

    (void)arg;
    char buffer[DATA_BUFFER_MAXSIZE] = {0};

    while(mq_readdata_thread_runtag) {
        mq_descriptor mq_des[MAX_MQ_DESCRIPTORS] = {0};
        pthread_rwlock_wrlock(&mq_des_rwlock);
        memcpy(mq_des, mq_descriptors, sizeof (mq_descriptor) * MAX_MQ_DESCRIPTORS);
        pthread_rwlock_unlock(&mq_des_rwlock);

        for (int i = 0; i < MAX_MQ_DESCRIPTORS; ++i) {
            if (0 <= mq_des[i].msgid && 1 == mq_des[i].valid) {
                memset(buffer, 0, sizeof (buffer)); // 清除上一次数据缓存
#ifdef __linux__
                ssize_t size = msgrcv(mq_des[i].msgid, buffer, DATA_BUFFER_MAXSIZE, 0, IPC_NOWAIT);
                if (0 < size) {
                    // 数据回调
                    if (mq_readdata_callback_pfn) {
                        mq_readdata_callback_pfn(mq_des[i].msgid, buffer, (size_t)size, mq_call_context);
                    }
                }
#else
                LPVOID hBuffer = (LPVOID)(mq_des[i].msgid);
                // windows下使用共享内存，无法获取实际读取了多少数据
                // 共享内存头4字节存储实际数据长度，后面为实际数据
                // 读取数据长度
                DWORD dwLen = 0;
                CopyMemory(&dwLen, hBuffer, sizeof(DWORD)); // 读取数据长度
                char* hDataHeader = (char*)hBuffer + sizeof(DWORD); // 获取数据起始位置
                // 读取数据
                CopyMemory(buffer, hDataHeader, dwLen/*sizeof(buffer) - sizeof(DWORD)*/);
                // 数据回调
                if (mq_readdata_callback_pfn && 0 < dwLen) {
                    mq_readdata_callback_pfn(mq_des[i].msgid, buffer, (size_t)dwLen, mq_call_context);
                }
#endif
            }
        }

        select_sleep(5); // 让出CPU时间
    }

    return NULL;
}

/******************************************************************
 * 功能：创建并打开消息队列，注：多进程通信时，消息队列需遵循谁创建谁销毁的原则
 * 输入参数：
 *      key：通信双方约定的相同键值
 * 返回值：
 *      int：消息队列的fd信息，小于0 -- 失败，大于等于0 -- 成功
 * ****************************************************************/
int msgqueue_create_and_open_by_key(int key)
{
    return create_or_open_mq_by_key(key, 0);
}

/******************************************************************
 * 功能：创建并打开消息队列，注：多进程通信时，消息队列需遵循谁创建谁销毁的原则
 * 输入参数：
 *      mq_path：通信双方约定的相同路径信息，路径未存在时主动创建，销毁队列时删除路径
 *              注：路径需要有读写访问权限
 *      mq_proj_id：通信双方约定的key信息，通常为一个int32的数
 * 返回值：
 *      int：消息队列的fd信息，小于0 -- 失败，大于等于0 -- 成功
 * ****************************************************************/
int msgqueue_create_and_open(const char* mq_path, int mq_proj_id)
{
    return create_or_open_mq(mq_path, mq_proj_id, 0);
}

/******************************************************************
 * 功能：打开一个已经存在的消息队列；
 *      一般情况下进程A、B通信时，可以由A创建并打开，B使用时只打开即可
 * 输入参数：
 *      key：通信双方约定的相同键值
 * 返回值：
 *      int：消息队列的fd信息，小于0 -- 失败，大于等于0 -- 成功
 * ****************************************************************/
int msgqueue_open_by_key(int key)
{
    return create_or_open_mq_by_key(key, 1);
}

/******************************************************************
 * 功能：打开一个已经存在的消息队列；
 *      一般情况下进程A、B通信时，可以由A创建并打开，B使用时只打开即可
 * 输入参数：
 *      mq_path：通信双方约定的相同路径信息，前提是路径先要被创建
 *      mq_proj_id：通信双方约定的key信息，通常为一个int32的数
 * 返回值：
 *      int：消息队列的fd信息，小于0 --失败，大于等于0 -- 成功
 * ****************************************************************/
int msgqueue_open(const char* mq_path, int mq_proj_id)
{
    return create_or_open_mq(mq_path, mq_proj_id, 1);
}

/******************************************************************
 * 功能：关闭一个已经存在的消息队列，注：这里主要是设置消息队列的无效标识
 * 输入参数：
 *      fd：消息队列的fd
 * ****************************************************************/
void msgqueue_close(int fd)
{
    close_or_destroy_mq(fd, 1);
    return;
}

/******************************************************************
 * 功能：关闭并销毁消息队列，注：多进程通信时，消息队列需遵循谁创建谁销毁的原则
 * 输入参数：
 *      fd：消息队列的fd
 * ****************************************************************/
void msgqueue_close_and_destroy(int fd)
{
    close_or_destroy_mq(fd, 0);
    return;
}

/******************************************************************
 * 功能：从消息队列中读取数据
 * 输入参数：
 *      fd：消息队列的fd
 *      buffer：读取数据的缓存空间
 *      buf_size：数据缓存空间的大小
 * 返回值：
 *      int：读取结果大于，0 -- 成功，具体读取数据的大小，-1 -- 不成功
 * ****************************************************************/
int msgqueue_read(int fd, void* buffer, size_t buf_size)
{
    if (!buffer || 0 >= buf_size || DATA_BUFFER_MAXSIZE < buf_size) {
        return -1;
    }

    // 先检测消息队列id的有效性
    int idx = get_mq_index(fd);
    if (-1 == idx || MAX_MQ_DESCRIPTORS <= idx) {
        printf("msgqueue_read: invaild message id (%d).\r\n", fd);
        return -1;
    }

#ifdef __linux__
    // 使用非阻塞模式，读取队列第一个消息
    ssize_t size = msgrcv(fd, buffer, buf_size, 0, IPC_NOWAIT);
#if 0
    // 不用提示消息
    if (0 > size) {
        perror("msgqueue_read: msgrcv failed.\r\n");
        return -1;
    }
#endif
    return (int)size;
#else
    // windows下使用共享内存，无法获取实际读取了多少数据
    // 需要先读取数据长度，再获取实际数据
    LPVOID hBuffer = (LPVOID)(fd);
    // 读取数据长度
    DWORD dwLen = 0;
    CopyMemory(&dwLen, hBuffer, sizeof(DWORD)); // 读取数据长度
    char* hDataHeader = (char*)hBuffer + sizeof(DWORD); // 获取数据起始位置

    if (buf_size >= dwLen) {
        // 读取数据
        CopyMemory(buffer, hDataHeader, dwLen);
    }

    return (int)dwLen;
#endif
}

/******************************************************************
 * 功能：把数据写入消息队列中
 * 输入参数：
 *      fd：消息队列的fd
 *      buffer：待写入数据的缓存空间
 *      size：数据大小
 * 返回值：
 *      int：写入结果，0 -- 成功，非0 -- 不成功
 * ****************************************************************/
int msgqueue_write(int fd, void* buffer, size_t size)
{
    if (!buffer || DATA_BUFFER_MAXSIZE < size) {
        return -1;
    }

    // 先检测消息队列id的有效性
    int idx = get_mq_index(fd);
    if (-1 == idx || MAX_MQ_DESCRIPTORS <= idx) {
        printf("msgqueue_write: invaild message id (%d).\r\n", fd);
        return -1;
    }

#ifdef __linux__
    // 使用非阻塞模式，写入消息
    if (0 > msgsnd(fd, buffer, size, IPC_NOWAIT)) {
        // msgsnd在IPC_NOWAIT模式下返回值为0表示发送成功
        perror("msgqueue_write: msgsnd failed.\r\n");
        return -1;
    }

    return 0;
#else
    // windows下使用共享内存，无法获取实际读取了多少数据
    // 需要写入数据长度，再写入数据
    LPVOID hBuffer = (LPVOID)(fd);
    // 写入数据长度
    DWORD dwLen = (DWORD)size;
    CopyMemory(hBuffer, &dwLen, sizeof (DWORD)); // 写入数据长度
    // 写入数据
    char* hDataHeader = (char*)hBuffer + sizeof(DWORD); // 获取数据起始位置
    CopyMemory(hDataHeader, buffer, dwLen); // 写入数据
    return 0;
#endif
}

/******************************************************************
 * 功能：设置消息队列处理回调函数
 * 输出参数：
 *      pfn：回调函数
 *      context：调用的上下文实例
 * ****************************************************************/
void msgqueue_set_readdata_callback(pfn_mq_readdata_callback pfn, void *context)
{
    mq_readdata_callback_pfn = pfn;
    mq_call_context = context;
    return;
}

/******************************************************************
 * 功能：开始消息队列自动读取数据的线程
 * ****************************************************************/
void msgqueue_loop_start()
{
    if (1 == mq_readdata_thread_runtag) {
        perror("msgqueue_loop_start: it is already started, please call stop first.\r\n");
        return;
    }

    mq_readdata_thread_runtag = 1;
    if (0 != pthread_create(&mq_readdata_thread_hd, NULL, pfn_readdata_mq, NULL)) {
        perror("msgqueue_loop_start: create thread failed.\r\n");
        return;
    }

    return;
}

/******************************************************************
 * 功能：停止消息队列自动读取数据的线程
 * ****************************************************************/
void msgqueue_loop_stop()
{
    if (1 == mq_readdata_thread_runtag) {
        mq_readdata_thread_runtag = 0; // 设置停止标识
        pthread_join(mq_readdata_thread_hd, NULL);
        mq_readdata_thread_hd = 0;
    }

    return;
}

